/***
 * Copyright (c) 2021-2031 murenchao
 * fig is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *       http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */
package cool.taomu.software.fig.mqtt.broker

import com.google.common.collect.HashBasedTable
import com.google.common.collect.Table
import cool.taomu.software.fig.mqtt.broker.entity.ClientSessionEntity
import cool.taomu.software.fig.mqtt.broker.entity.MessageEntity
import cool.taomu.software.fig.mqtt.broker.entity.WillEntity
import cool.taomus.software.fig.cache.FigCache
import java.util.HashMap
import java.util.HashSet
import java.util.Set
import java.util.concurrent.BlockingQueue
import java.util.concurrent.LinkedBlockingQueue

class MQTTCache {
    enum CacheType {
        RETAIN,
        SESSION,
        QoS2,
        WILL
    }

    val static map = new HashMap<String, BlockingQueue<MessageEntity>>;
    val static Table<String, String, Set<Object>> clientInfo = HashBasedTable.create();
    val static instance = new MQTTCache();
    val static cache = FigCache.instance;

    private new() {
    }

    def static getInstance() {
        return instance;
    }

    def store(String key, MessageEntity value) {
        if (value !== null) {
            if (map.containsKey(key)) {
                map.get(key).put(value);
            } else {
                map.put(key, new LinkedBlockingQueue)
                map.get(key).put(value);
            }
        }
    }

    def storeQoS2Message(String clientId, MessageEntity value) {
        if (clientInfo.contains(clientId, CacheType.QoS2.name)) {
            var retains = clientInfo.get(clientId, CacheType.QoS2.name);
            retains.add(value);
        } else {
            var retains = new HashSet();
            retains.add(value);
            clientInfo.put(clientId, CacheType.QoS2.name, retains);
        }
    }

    def loadQoS2Message(String clientId) {
        val results = new HashSet<Object>()
        clientInfo.column(CacheType.QoS2.name).forEach [ p1, msg, p3 |
            results.addAll(msg);
        ]
        return results;
    }

    def clearQoS2Message(String clientId, MessageEntity value) {
        return clientInfo.get(clientId, CacheType.QoS2.name).remove(value);
    }

    def loadMessage(String key) {
        if (map.containsKey(key)) {
            return map.get(key).poll
        }
        return null;
    }

    def storeSession(String id, ClientSessionEntity value) {
        var session = new HashSet();
        session.add(value);
        clientInfo.put(id, CacheType.SESSION.name, session);
    }

    def loadSession(String clientId) {
        if (clientInfo.contains(clientId, CacheType.SESSION.name)) {
            var results = clientInfo.get(clientId, CacheType.SESSION.name);
            if (results !== null && results.size > 0) {
                return results.get(0) as ClientSessionEntity;
            }
        }
        return null;
    }

    def clearSession(String clientId) {
        if (clientInfo.contains(clientId, CacheType.SESSION.name)) {
            clientInfo.remove(clientId, CacheType.SESSION.name);
        }
    }

    def storeWill(String clientId, WillEntity value) {
        var will = new HashSet();
        will.add(value);
        clientInfo.put(clientId, CacheType.WILL.name, will);
    }

    def loadWill(String clientId) {
        if (clientInfo.contains(clientId, CacheType.WILL.name)) {
            var results = clientInfo.get(clientId, CacheType.WILL.name);
            if (results !== null && results.size > 0) {
                return results.get(0) as WillEntity;
            }
        }
        return null;
    }

    def clearWill(String clientId) {
        if (clientInfo.contains(clientId, CacheType.WILL.name)) {
            clientInfo.remove(clientId, CacheType.WILL.name);
        }
    }

    def storeRetain(String senderId, MessageEntity value) {
        var result = cache.get(CacheType.RETAIN.name) as HashMap<String, Set<MessageEntity>>;
        if (result === null) {
            var retainMap = new HashMap<String, Set<MessageEntity>>();
            var retains = new HashSet<MessageEntity>();
            retains.add(value);
            retainMap.put(senderId, retains);
            cache.store(CacheType.RETAIN.name, retainMap);
        } else {
            if (result.containsKey(senderId)) {
                result.get(senderId).add(value);
            } else {
                var retains = new HashSet<MessageEntity>();
                retains.add(value);
                result.put(senderId, retains);
                cache.store(CacheType.RETAIN.name, result);
            }
        }
    }

    def clearRetainBySenderId(String senderId) {
        var result = cache.get(CacheType.RETAIN.name) as HashMap<String, Set<MessageEntity>>;
        if (result !== null) {
            result.remove(senderId);
        }
    }
    
    def refreshRetain(){
        cache.refresh(CacheType.RETAIN.name);
    }

    def loadRetain() {
        var result = cache.get(CacheType.RETAIN.name) as HashMap<String, Set<MessageEntity>>;
        if (result !== null) {
            val results = new HashSet<MessageEntity>()
            result.values.filterNull.forEach[res|results.addAll(res)]
            return results;
        }
        return null;
    }

    def static void main(String[] args) {
        var m = new MessageEntity();
        m.topic = "1";
        MQTTCache.getInstance.storeRetain("1", m);
        m = new MessageEntity();
        m.topic = "2";
        MQTTCache.getInstance.storeRetain("1", m);
        m = new MessageEntity();
        m.topic = "3";
        MQTTCache.getInstance.storeRetain("2", m);
        println(MQTTCache.getInstance.loadRetain());
    }

}
